package org.leiyang.mq;

import org.leiyang.common.entities.MsgOneToOne;
import org.leiyang.common.holder.MsgServerAndClientHolder;
import org.leiyang.common.redis.ReactiveRedisCache;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;

import java.util.Collections;
import java.util.Map;
import java.util.Objects;
import org.leiyang.repository.MsgOneToOneRepository;

import static org.leiyang.common.util.Constants.MSG_ID_KEY;
import static org.leiyang.common.util.Constants.OFFLINE_MSGS_KEY;

/**
 * 来自客户端的新消息的消费处理
 */
@Component("newMsgConsumer")
public class NewMsgConsumer implements StreamListener<String, MapRecord<String, String, String>> {
    private final Logger logger = LoggerFactory.getLogger(getClass());
    public static final String topicName ="new_msg_topic";
    public static final String groupId = "new_msg_group_1";
    @Value("${newMsg.consumer.id}")
    private String newMsgConsumerId;

    private final ReactiveRedisCache redisCache;
    private final MsgOneToOneRepository msgOneToOneRepository;
    private final MsgServerAndClientHolder msgServerAndClientHolder;
    public NewMsgConsumer(ReactiveRedisCache redisCache, MsgOneToOneRepository msgOneToOneRepository,
                          MsgServerAndClientHolder msgServerAndClientHolder) {
        this.redisCache = redisCache;
        this.msgOneToOneRepository = msgOneToOneRepository;
        this.msgServerAndClientHolder = msgServerAndClientHolder;
    }

    /**
     * 1、将收到的新消息入库DB
     * 2、确认消息发送的用户是否在线，如果离线，则将该消息存储到redis队列里；如果在线，则将该消息回发到MQ里
     * @param message never {@literal null}.
     */
    @Override
    public void onMessage(MapRecord<String, String, String> message) {
        //业务处理
        logger.info("收到来自客户端的新信息[NewMsg]: {}", message);
        try {
            Map<String, String> msgMap = message.getValue();
            MsgOneToOne msg = MsgOneToOne.convertFromMap(msgMap);
            if (Objects.equals(msg.getFromUserId(), -1L)) {
                logger.error("消费者({})消费到的MsgOneToOne无效: {}", newMsgConsumerId, msg);
                return;
            }
            redisCache.increaseValue(MSG_ID_KEY, 1L)
                .map(msgId -> {
                    msg.setId(msgId);
                    return msg;
                })
                .flatMap(msgOneToOne -> msgOneToOneRepository.addNewMsg(msgOneToOne.getId(), msgOneToOne.getFromUserId(), msgOneToOne.getToUserId(), msgOneToOne.getMsgType(),
                        msgOneToOne.getTextMsg(), msgOneToOne.getAvMsgUrl(), msgOneToOne.getImgWidth(), msgOneToOne.getImgHeight(), msgOneToOne.getSend(), msgOneToOne.getTime()))
                .flatMapMany(rowNum -> msgServerAndClientHolder.confirmUserOnline(msg.getToUserId()))
                .hasElements()
                .flatMap(online -> {
                    //在线的话，将单聊消息回发到MQ
                    if(online) {
                        msgMap.put("id", msg.getId().toString());
                        return redisCache.createStreamOperations()
                                .add(OnlineMsgConsumer.topicName, msgMap)
                                .map(recordId -> true);
                    } else {
                        //离线的话，将该消息视为离线消息存储到redis队列里
                        return redisCache.setCacheSet(OFFLINE_MSGS_KEY + msg.getToUserId(), Collections.singleton(msg))
                                .map(num -> true);
                    }
                }).subscribe();
        } catch (Exception e) {
            logger.error("消费到MsgOneToOne:{} 并保存到DB时出现异常, ", message.getValue(), e);
        } finally {
            //始终会消息应答
            redisCache.createStreamOperations().acknowledge(topicName, groupId, message.getId());
        }
    }

    //todo: 启动定时任务不断监听pending列表，处理死信消息
}
